package top.ivansong.cxf.registry;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.cxf.jaxrs.client.JAXRSClientFactory;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs.Ids;
import org.codehaus.jackson.jaxrs.JacksonJaxbJsonProvider;
import org.springframework.beans.factory.DisposableBean;
import top.ivansong.cxf.registry.api.ServiceBeanInfo;
import top.ivansong.cxf.registry.api.ServiceNode;
import top.ivansong.cxf.registry.api.ServiceRegistry;
import top.ivansong.cxf.registry.util.ServiceBeanSerializeUntil;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ZookeeperRegistry implements ServiceRegistry, DisposableBean {

    private final Log log = LogFactory.getLog(ZookeeperRegistry.class);
    private CuratorFramework client;

    private JacksonJaxbJsonProvider provider = new JacksonJaxbJsonProvider();

    private ExecutorService threedPool;

    private ConcurrentMap<String, Map<String, List<ServiceNode>>> serviceNodes = new ConcurrentHashMap<>();

    private ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();

    private Set<String> cusumeServiceCache = new HashSet<>();

    private int listenerSize = 5;

    private static PathChildrenCacheEvent.Type[] types = { PathChildrenCacheEvent.Type.CHILD_REMOVED, PathChildrenCacheEvent.Type.CHILD_ADDED, PathChildrenCacheEvent.Type.CHILD_UPDATED};
    private static Set<PathChildrenCacheEvent.Type> listenedType = new HashSet<>();

    static {
        listenedType.addAll(Arrays.asList(types));
    }

    public ZookeeperRegistry(String connectString, int timeOut) {
        client = CuratorFrameworkFactory.builder()
                .connectionTimeoutMs(timeOut)
                .connectString(connectString)
                .retryPolicy(new ExponentialBackoffRetry(timeOut, 3))
                .build();
        client.start();
        threedPool = Executors.newFixedThreadPool(listenerSize);
    }

    @Override
    public void registService(ServiceBeanInfo bean) {
        String nodeInfo = ServiceBeanSerializeUntil.encoded(bean);
        try {
            client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).withACL(Ids.OPEN_ACL_UNSAFE)
                    .forPath(getServiceProviderPath(bean.getClassName()) + "/" + ServiceBeanSerializeUntil.encodedUrl(bean.getUrl()), nodeInfo.getBytes());
        } catch (Exception e) {
            log.warn("node is registed.", e);
        }
    }

    @Override
    public void subscribeService(final String serviceName) {
        boolean notExsit = cusumeServiceCache.add(serviceName);
        if (!notExsit) return;

        PathChildrenCache node = new PathChildrenCache(client, getServiceProviderPath(serviceName), true);
        try {
            node.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework client, final PathChildrenCacheEvent event) throws Exception {
                    threedPool.execute(new Runnable() {
                        @Override
                        public void run() {
                            refreshServiceNodes(serviceName, event);
                        }
                    });
                }
            });
            node.start();
            Map<String, List<ServiceNode>> map = serviceNodes.get(serviceName);
            List<ChildData> childs = node.getCurrentData();
            for(ChildData child : childs) {
                try {
                    String data = new String(child.getData());
                    ServiceBeanInfo beaInfo = ServiceBeanSerializeUntil.decode(data);
                    ServiceNode serviceNode = converInfoToNode(beaInfo);
                    if (map == null) {
                        map = new HashMap<>();
                    }
                    map.put(beaInfo.getUrl(), Arrays.asList(serviceNode));
                    serviceNodes.put(serviceName, map);
                } catch (Exception e1) {
                    log.error("subscribeService error.", e1);
                }
            }
        } catch (Exception e) {
            log.error("subscribeService error.", e);
        }
    }


    @Override
    public List<ServiceNode> getServiceNodes(String serviceName) {
        rwLock.readLock().lock();
        try {
            Map<String, List<ServiceNode>>  serviceMap = serviceNodes.get(serviceName);
            if (serviceMap == null || serviceMap.isEmpty()) {
                return Collections.emptyList();
            }

            Collection<List<ServiceNode>> values = serviceMap.values();
            List<ServiceNode> res = new ArrayList<>();
            for (List<ServiceNode> node : values) {
                res.addAll(node);
            }
            return res;
        } finally {
            rwLock.readLock().unlock();
        }
    }

    @Override
    public void destroy() throws Exception {
        threedPool.shutdown();
        client.close();
    }

    private void refreshServiceNodes(String serviceName, PathChildrenCacheEvent event) {
        if (event == null || !listenedType.contains(event.getType())) {
            return;
        }

        rwLock.writeLock().lock();
        try {
            Map<String, List<ServiceNode>> map = serviceNodes.get(serviceName);
            String data = new String(event.getData().getData());
            ServiceBeanInfo beaInfo = ServiceBeanSerializeUntil.decode(data);
            if (event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
                if (map != null) {
                    map.remove(beaInfo.getUrl());
                }
            }
            if (event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED) {
                ServiceNode node = converInfoToNode(beaInfo);
                if (map == null) {
                    map = new HashMap<>();
                }
                map.put(beaInfo.getUrl(), Arrays.asList(node));
                serviceNodes.put(serviceName, map);
            }

            if (event.getType() == PathChildrenCacheEvent.Type.CHILD_UPDATED) {
                if (map == null) {
                    map = new HashMap<>();
                    ServiceNode nodeUp = converInfoToNode(beaInfo);
                    map.put(beaInfo.getUrl(), Arrays.asList(nodeUp));
                    serviceNodes.put(serviceName, map);
                }
            }
        } catch (Exception e) {
            log.error("refreshServiceNodes error.", e);
        } finally {
            rwLock.writeLock().unlock();
        }
    }

    private ServiceNode converInfoToNode(ServiceBeanInfo beaInfo) throws ClassNotFoundException {
        ServiceNode node = new ServiceNode();
        node.setAddress(beaInfo.getUrl());
        node.setServiceInfo(beaInfo);
        String baseAddress = beaInfo.getUrl() + (beaInfo.getUrlMapping().startsWith("/") ? beaInfo.getUrlMapping() : "/" + beaInfo.getUrlMapping());
        Object service = JAXRSClientFactory.create(baseAddress, Class.forName(beaInfo.getClassName()), Arrays.asList(provider));
        node.setService(service);

        return node;
    }

    private String getServiceProviderPath(String className) {
        return "/cxf/" + className + "/provider";
    }
}